package com.atguigu.orderpay_detect

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
  * 2
  * @author: yangShen
  * @Description: 使用状态来实现订单超时状态的处理
  *              问题：不能做到来一条马上输出，要等时间戳 过15分钟
  *
  *              解决：来一条处理一条，正常的放主流，不正常的放侧输出流，
  * @Date: 2020/5/11 15:55 
  */
object OrderTimeoutWithoutCep {

  //定义侧输出流标签
  val orderTimeoutOutputTag = new OutputTag[OrderResult]("orderTimeout")

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    environment.setParallelism(1)

    //1.读取订单数据
    val resource = getClass.getResource("/OrderLog.csv")
//        val orderEventStream = environment.readTextFile(resource.getPath) //文件读取
    val orderEventStream = environment.socketTextStream("10.16.26.16", 7777)  //流式读取：nc -lk 7777
      .map(data => {
      val dataArray = data.split(",")
      OrderEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
    })
      //设置时间戳
      //方式一(简单实现)：升序数据，不设置延时 到点就发车，从字段中抽取
      .assignAscendingTimestamps(_.eventTime * 1000L)
      .keyBy(_.orderId)

    //定义process function 处理超时检测
//    val timeoutWarningStream = orderEventStream.process(new OrderTimeoutWarning())

    //匹配各种类型
    val orderResultStream = orderEventStream.process(new OrderPayMatch())


//    timeoutWarningStream.print()
    orderResultStream.print("payed")
    //把所有的异常情况 都输出到侧输出流
    orderResultStream.getSideOutput(orderTimeoutOutputTag).print("timeout")


    environment.execute("order timeout without cep job")
  }

  class OrderPayMatch() extends KeyedProcessFunction[Long, OrderEvent, OrderResult]{

    //保存pay是否来过的状态
    lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("ispayed-state", classOf[Boolean]))
    //保存定时器的时间戳为状态：为了删除定时器，如果定时器的时间戳是0表示create没来
    lazy val timerState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timer-state", classOf[Long]))

    override def processElement(value: OrderEvent, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, out: Collector[OrderResult]): Unit = {

      //先读取状态
      val isPayed = isPayedState.value()
      val timerTs = timerState.value()

      //根据事件的类型进行分类判断，做不同的逻辑；  如果先来的pay 15分钟之后来的create，这种情况在此不考虑
      if (value.eventType == "create"){
        // 1.如果是create事件，判断pay是否来过
        if (isPayed){
          // 1.1如果pay已经来过，匹配成功，输出主流数据，清空状态
          out.collect( OrderResult(value.orderId, "payed successfully"))
          ctx.timerService().deleteEventTimeTimer(timerTs)
        }else {
          // 1.2如果pay没有来过，注册定时器等待pay的到来，注册事件戳以value事件时间为起点 15分钟之后
          val ts = value.eventTime * 1000L + 15 * 60 * 1000L
          ctx.timerService().registerEventTimeTimer(ts)
          timerState.update(ts)
        }

      }else if (value.eventType == "pay"){
        // 2.如果是pay事件，要判断create是否来过，用timer表示
        if (timerTs > 0){
          // 2.1如果有定时器，说明已经有create来过
          //继续判断，是否超过了timeout时间
          if (value.eventTime * 1000L < timerTs){
            // 2.1.1如果定时器时间还没到，那么输出成功匹配
            out.collect(OrderResult(value.orderId, "payed successfully"))   //主流
          }else {
            //2.1.2如果当前pay的时间已经超时，那么输出到侧输出流
            ctx.output(orderTimeoutOutputTag, OrderResult(value.orderId, "payed but already timeout 15 minutes"))   //侧输出流
          }
          //输出结束，清空状态
          ctx.timerService().deleteEventTimeTimer(timerTs)
          isPayedState.clear()
          timerState.clear()
        }else {
          // 2.2说明pay先到了，更新状态，注册定时器等待create
          isPayedState.update(true)
          //定时器时间戳：等到水位到了watermark=value.event*1000L 时看create来没来，如果没来则create丢失，因为create应该在pay之前
          ctx.timerService().registerEventTimeTimer(value.eventTime*1000L)
          timerState.update(value.eventTime * 1000L)
        }
      }

    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {
      //根据状态的值判断哪个数据没来
      if(isPayedState.value()){
        //如果为true，表示pay先到了，没有等到create，比较少见说明create丢失了
        ctx.output(orderTimeoutOutputTag, OrderResult(ctx.getCurrentKey, "already payed but not fount create"))
      }else {
        //表示create到了，没有等到pay-----超时
        ctx.output(orderTimeoutOutputTag, OrderResult(ctx.getCurrentKey, "order timeout"))
      }
      //
      isPayedState.clear()
      timerState.clear()
    }
  }
}

//实现自定义的处理函数
class OrderTimeoutWarning extends KeyedProcessFunction[Long, OrderEvent, OrderResult]{

  //保存pay是否来过的状态
  lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("ispayed-state", classOf[Boolean]))

  //前提条件是，一个订单一个create一个pay,如果多个create同一个订单会注册多个定时器，生产中应该是一个订单一个create一个pay
  //如果是pay先来，也要等create到来，不然没有定时器
  override def processElement(value: OrderEvent, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, out: Collector[OrderResult]): Unit = {
    //先出去状态标识
    val isPayed = isPayedState.value()

    if (value.eventType == "create" && !isPayed){
      //如果碰到了create事件，并且pay没有来过，就注册定时器开始等待
      //事件时间 + 15分钟
      ctx.timerService().registerEventTimeTimer(value.eventTime * 1000L + 15 * 60 * 1000L)
    }else if(value.eventType == "pay"){
      //如果是pay事件，直接把状态改为true
      isPayedState.update(true)
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {
    //判断isPayedState是否为true
    val isPayed = isPayedState.value()
    if (isPayed){
      out.collect( OrderResult(ctx.getCurrentKey, "order payed successfully"))
    }else{
      out.collect(OrderResult(ctx.getCurrentKey, "order timeout"))
    }

    //清空状态
    isPayedState.clear()
  }

}
